package com.zhike.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhike.bean.TempBean;
import com.zhike.function.BeanToJsonFunction;
import com.zhike.utils.DateFormatUtil;
import com.zhike.utils.DorisUtils;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Random;

/**
 * @Author: zhike
 * @CreateTime: 2024/2/26 16:07
 * @Description: 读取电动模式行驶日志，混动模式行驶日志，新能源燃料模式行驶日志，充电桩充电日志，电机行驶日志
 * 创建流计算各车的电机温度和电池温度
 */
public class DwsTempWindow {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setParallelism(2);
        Random random = new Random();

        //2.读取car_electricity_log日志创建流并转换为TempBean对象
        SingleOutputStreamOperator<TempBean> eleDS = env.fromSource(KafkaUtils.getKafkaSource("car_electricity_log", "dws_temp_window")
                        , WatermarkStrategy.noWatermarks(), "car_electricity_log_source")
                .map(JSON::parseObject)
                .map(DwsTempWindow::getTempBean);

        //3.读取car_hybrid_log日志创建流并转换为TempBean对象
        SingleOutputStreamOperator<TempBean> hybridDS = env.fromSource(KafkaUtils.getKafkaSource("car_hybrid_log", "dws_temp_window")
                        , WatermarkStrategy.noWatermarks(), "car_hybrid_log_source")
                .map(JSON::parseObject)
                .map(DwsTempWindow::getTempBean);

        //4.读取car_fuel_log日志创建流并转换为TempBean对象
        SingleOutputStreamOperator<TempBean> fuelDS = env.fromSource(KafkaUtils.getKafkaSource("car_fuel_log", "dws_temp_window")
                        , WatermarkStrategy.noWatermarks(), "car_fuel_log_source")
                .map(JSON::parseObject)
                .map((MapFunction<JSONObject, TempBean>) DwsTempWindow::getTempBean);

        //5.读取car_parking_charging_log日志创建流并转换为TempBean对象
        SingleOutputStreamOperator<TempBean> parkingChargingDS = env.fromSource(KafkaUtils.getKafkaSource("car_parking_charging_log", "dws_temp_window")
                        , WatermarkStrategy.noWatermarks(), "car_parking_charging_log_source")
                .map(JSON::parseObject)
                .map(DwsTempWindow::getTempBean);

        //6.读取car_motor_log日志创建流并转换为TempBean对象
        SingleOutputStreamOperator<TempBean> motorDS = env.fromSource(KafkaUtils.getKafkaSource("car_motor_log", "dws_temp_window"),
                        WatermarkStrategy.noWatermarks(), "car_motor_log_source")
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, TempBean>() {
                    @Override
                    public TempBean map(JSONObject value) throws Exception {
                        //获取vin
                        String vin = value.getString("vin");
                        //获取当前日志的电机温度
                        Integer temp = value.getInteger("temperature");
                        int tmp = random.nextInt(11)+50;
                        temp=temp-tmp;
                        //获取当前日志的电机控制器温度
                        Integer conTemp = value.getInteger("controller_temperature");
                        int i = random.nextInt(11)+40;
                        conTemp=conTemp-i;
                        //获取时间戳
                        Long ts = value.getLong("timestamp");
                        return TempBean.builder()
                                .vin(vin)
                                .maxTempMotor(temp)
                                .avgTempMotorCount(1)
                                .maxTempMotorController(conTemp)
                                .avgTempMotorControllerCount(1)
                                .totalTempMotor(temp)
                                .totalTempMotorController(conTemp)
                                .ts(ts)
                                .build();
                    }
                });

        //7.将多流合并
        DataStream<TempBean> unionDS = eleDS.union(hybridDS, fuelDS, parkingChargingDS, motorDS);

        //8.设置水位线
        SingleOutputStreamOperator<TempBean> wMDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TempBean>forBoundedOutOfOrderness(Duration.ofSeconds(30L))
                .withTimestampAssigner(new SerializableTimestampAssigner<TempBean>() {
                    @Override
                    public long extractTimestamp(TempBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        //8.按照vin分组，开窗，聚合
        SingleOutputStreamOperator<TempBean> reduceDS = wMDS.keyBy(new KeySelector<TempBean, String>() {
                    @Override
                    public String getKey(TempBean value) throws Exception {
                        return value.getVin();
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(60L)))
                .reduce(new ReduceFunction<TempBean>() {
                    @Override
                    public TempBean reduce(TempBean value1, TempBean value2) throws Exception {
                        value1.setMaxTempMotor(Math.max(value1.getMaxTempMotor(), value2.getMaxTempMotor()));
                        value1.setTotalTempMotor(value1.getTotalTempMotor() + value2.getTotalTempMotor());
                        value1.setAvgTempMotorCount(value1.getAvgTempMotorCount() + value2.getAvgTempMotorCount());
                        value1.setMaxTempMotorController(Math.max(value1.getMaxTempMotorController(), value2.getMaxTempMotorController()));
                        value1.setTotalTempMotorController(value1.getTotalTempMotorController() + value2.getTotalTempMotorController());
                        value1.setAvgTempMotorControllerCount(value1.getAvgTempMotorControllerCount() + value2.getAvgTempMotorControllerCount());
                        value1.setMaxTempBattery(Math.max(value1.getMaxTempBattery(), value2.getMaxTempBattery()));
                        value1.setBatteryTempAbnormalCount(value1.getBatteryTempAbnormalCount() + value2.getBatteryTempAbnormalCount());
                        return value1;
                    }
                }, new WindowFunction<TempBean, TempBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TempBean> input, Collector<TempBean> out) throws Exception {
                        TempBean bean = input.iterator().next();
                        bean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        bean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        bean.setWindowDate(DateFormatUtil.toDate(window.getStart()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });
        reduceDS.print();

        //9.将结果写入Doris表
        SingleOutputStreamOperator<String> resultDS = reduceDS.map(new MapFunction<TempBean, String>() {
            @Override
            public String map(TempBean value) throws Exception {
                return BeanToJsonFunction.getString(value);
            }
        });

        resultDS.sinkTo(DorisUtils.getDorisSink("dws_temp_window"));

        //10.启动任务
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static TempBean getTempBean(JSONObject value) {
        //获取该日志的最高电池温度
        Integer maxTemperature = value.getInteger("max_temperature");
        //获取vin
        String vin = value.getString("vin");
        //获取时间戳
        Long ts = value.getLong("timestamp");
        return TempBean.builder()
                .vin(vin)
                .maxTempBattery(maxTemperature)
                .batteryTempAbnormalCount(maxTemperature > 500 ? 1 : 0)
                .ts(ts)
                .build();
    }
}
